package com.atguigu.gmall.realtime.app

import java.lang

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.gmall.realtime.bean.{PageActionLog, PageLog}
import com.atguigu.gmall.realtime.util.{MyKafkaSink, MyKafkaUtil, OffsetManager}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object BaseLogApp {

  //1  消费kafka主题
  //2 转换结构
  //3  分流  ：
  //     启动主题 ， DWD_START_LOG  对应pageLog
  //     页面访问主题  DWD_PAGE_LOG
  //     页面动作主题 DWD_PAGE_ACTION
  //     页面曝光主题  DWD_PAGE_DISPLAY
  //     错误主题    DWD_ERROR_INFO

  // 语义： 至少一次消费
  def main(args: Array[String]): Unit = {
    //1  消费kafka主题
      val sparkConf: SparkConf = new SparkConf().setAppName("base_log_app").setMaster("local[4]")
       val ssc = new StreamingContext(sparkConf,Seconds(5))

      val topic="ODS_BASE_LOG"
      val  groupId="base_log_group"

      val offsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic,groupId)

    // 加载流
      var inputDstream: InputDStream[ConsumerRecord[String, String]]=null
      if(offsetMap!=null && offsetMap.size>0){
          inputDstream  = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap ,groupId)
      }else{
          inputDstream  = MyKafkaUtil.getKafkaStream(topic,ssc ,groupId)

      }

    //  inputDstream.map(record=> record.value()) .print()
    //  inputDstream.map(_.value()).print(10000)

    // offset 3 在数据流转换前 提取偏移量结束点 ，存储在变量中
    var offsetRanges: Array[OffsetRange]=null  //driver
    val inputOffsetDstream: DStream[ConsumerRecord[String, String]] = inputDstream.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //driver?  driver executor?  //执行频率？ 每批次执行一次
      rdd
    }




    //  //2 转换结构
    //把kafka的消息转换为结构化对象 ：//1 通用的结构化对象 hashmap,jsonObject  2 专用的结构化对象 javabean \ case class
    val jsonObjDstream: DStream[JSONObject] = inputOffsetDstream.map { record =>
      val jsonString: String = record.value()
      val jsonObj: JSONObject = JSON.parseObject(jsonString)
      jsonObj
    }

    // 3 分流

        jsonObjDstream.foreachRDD{rdd=>
          rdd.foreachPartition { jsonObjItr => // 每个分区 每批次  // executor
            for (jsonObj <- jsonObjItr) {
              //1 提取页面访问主题的数据
              val commonObj: JSONObject = jsonObj.getJSONObject("common")
              val mid: String = commonObj.getString("mid")
              val uid: String = commonObj.getString("uid")
              val ar: String = commonObj.getString("ar")
              val ch: String = commonObj.getString("ch")
              val os: String = commonObj.getString("os")
              val md: String = commonObj.getString("md")
              val vc: String = commonObj.getString("vc")
              val isNew: String = commonObj.getString("is_new")

              val ts: lang.Long = jsonObj.getLong("ts")

              val pageObj: JSONObject = jsonObj.getJSONObject("page")
              if (pageObj != null && pageObj.size() > 0) {
                val pageId = pageObj.getString("page_id")
                val pageItem = pageObj.getString("item")
                val pageItemType = pageObj.getString("item_type")
                val lastPageId = pageObj.getString("last_page_id")
                val duringTime = pageObj.getLong("during_time")

                val pageLog = PageLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, ts)
                //发送到kafka指定主题中
                MyKafkaSink.send("DWD_PAGE_LOG", JSON.toJSONString(pageLog, new SerializeConfig(true)))
                //分流 页面动作主题
                val actionJsonArr: JSONArray = jsonObj.getJSONArray("actions")
                if (actionJsonArr != null && actionJsonArr.size() > 0) {
                  for (i <- 0 to actionJsonArr.size() - 1) {
                    val actionJsonObj: JSONObject = actionJsonArr.getJSONObject(i)
                    val action_id = actionJsonObj.getString("action_id")
                    val item = actionJsonObj.getString("item")
                    val item_type = actionJsonObj.getString("item_type")
                    val pageActionLog = PageActionLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, action_id, item, item_type, ts)
                    MyKafkaSink.send("DWD_PAGE_ACTION", JSON.toJSONString(pageActionLog, new SerializeConfig(true)))


                  }

                }
                // 分流 曝光主题 ( 自己完成 ，自己定义case class )
              }
              //  分流 启动信息主题 （自己完成，课件中有case class）
              //  分流 错误信息主题 （ 自己完成 ，不用转换直接把整个json发送到错误主题)


              //A 提交偏移量？   foreach里面    executor  每条数据执行一次
              // MyKafkaSink.flush()  // 不能每次send完都flush  这样等于同步写

            }
            //C 是 A和B都不对   要写在一个位置  既能保证是批量 同时还要是executor的位置。
            MyKafkaSink.flush()

          }
          //B 提交偏移量？   foreachRDD里面   driver ?  executor?  d     执行频率？一个批次执行一次
          // 提交偏移量

          //在提交偏移量之前 要进行flush操作 ：把缓冲的数据一起发送到kafka broker上
          // MyKafkaSink.flush() ?写在哪  //在driver中无法flush executor的缓冲区

          OffsetManager.saveOffset(topic,groupId,offsetRanges)

        }
      //C  提交偏移量？    foreachRDD外面   driver ?  executor ?  d     执行频率？ 只执行一次 启动应用程序时



       ssc.start()
       ssc.awaitTermination()



  }

}
